/*
 * Indiana University Community Grid Computing Lab Software License,Version 1.1
 *
 * Copyright (c) 2002 Community Grid Computing Lab, Indiana University.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *      "This product includes software developed by the Indiana University
 *       Community Grid Computing Lab (http://www.communitygrids.iu.edu/)."
 *    Alternately, this acknowledgment may appear in the software itself,
 *    if and wherever such third-party acknowledgments normally appear.
 *
 * 4. The names "Indiana Univeristy","Indiana Univeristy Pervasive Techonology
 *    Labs" and  "Indiana Univeristy Community Grid Computing Lab" must not be
 *    used to endorse or promote products derived from this software without
 *    prior written permission. For written permission, please contact
 *    http://www.communitygrids.iu.edu/.
 *
 * 5. Products derived from this software may not use "Indiana
 *    University" name nor may "Indiana University" appear in their name,
 *    without prior written permission of the Indiana University.
 *
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
 * INDIANA UNIVERSITY OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
 * OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
 * EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package cgl.sensorgrid.filters;


import java.util.*;
import cgl.narada.event.*;
import cgl.narada.matching.*;
import cgl.narada.service.*;
import cgl.narada.service.client.*;
import cgl.sensorgrid.common.*;
import org.doomdark.uuid.*;
import java.lang.reflect.*;

public class Filter implements NBEventListener {
    private Properties properties;
    private String nbServConfPath;
    private String subTopic;
    private String pubTopic;
    private String nbHost;
    private String nbPort;
    private String commType;
    private int entityId;
    private static int pubID = 0;
    private static Hashtable hash = new Hashtable();
    private Object entityObj = null;

    private ClientService clientService = null;
    private EventConsumer eventConsumer = null;
    private EventProducer eventProducer = null;
    private static boolean initialized = false;
    private static int count = 0;
    private Object syncObj = new Object();
    private Profile profile = null;
    long time = 0;
    private String uid;
    private UUIDGenerator uig = UUIDGenerator.getInstance();

//    private static Filter instance = new Filter();

//    public static Filter getInstance() {
//        return instance;
//    }


    /**
     *
     *
     */
    public Filter() {
        try {
            /* Enumeration enumer = hash.elements();
             System.out.println("\nHASH ELEMENTS");
             while (enumer.hasMoreElements()) {
                 Filter item = (Filter) enumer.nextElement();
                 System.out.println(item.getID());
             }

             System.out.println("\nHASH KEYS");
             Enumeration en = hash.keys();
             System.out.println("hash.size()=" + hash.size());
             while (en.hasMoreElements()) {
                 String item = (String) en.nextElement();
                 System.out.println(item);
             }
             */
            properties = PropertyFile.loadProperties("sensorgrid.properties");
            nbServConfPath = properties.getProperty("service.config.file");
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public void onEvent(NBEvent nbEvent) {}

    /**
     *
     *
     *
     * @param args String[]
     * @return boolean
     */
    public boolean initialize(String[] args) {
        if (args == null ||
            args.length == 0 ||
            args.length < 5) {
            System.err.println(
                    "Incorrect number of arguments in Filter::init()...");
            return false;
        }
        if (args[0] != null && !args[0].equals(""))
            this.commType = args[0];
        if (args[1] != null && !args[1].equals(""))
            this.nbHost = args[1];
        if (args[2] != null && !args[2].equals(""))
            this.nbPort = args[2];
        if (args[3] != null && !args[3].equals(""))
            this.subTopic = args[3];
        if (args[4] != null && !args[4].equals(""))
            this.pubTopic = args[4];

        if (subTopic != null && !subTopic.equals(""))
            this.initSubscriber();
        if (pubTopic != null && !pubTopic.equals(""))
            this.initPublisher();

        uid = uig.generateRandomBasedUUID().toString();
        hash.put(uid, this);
        System.out.println("Filter Hash Size = " + hash.size());

        System.out.println("\nFilter Connection Parameters:");
        System.out.println("  NB HOST   = " + nbHost);
        System.out.println("  NB PORT   = " + nbPort);
        System.out.println("  SUB TOPIC = " + subTopic);
        System.out.println("  PUB TOPIC = " + pubTopic);
        System.out.println("  FILTER ID = " + uid);

        return true;
    }

    /**
     *
     *
     *
     */
    public synchronized void getEntityId() {
        synchronized (syncObj) {
            int t1 = (int) (1000000 * Math.random());
            entityId = (int) (System.currentTimeMillis() % 1000000) + t1;
            if (pubID == entityId)
                entityId = (int) (entityId + t1);
            else
                pubID = entityId;

            entityId = Math.abs(entityId);
            System.out.println("\n--> Filter No = " + count++ +
                               "\n    Entity Id    = " + entityId);
            entityObj = new Integer(entityId);
        }
    }

    /**
     *
     *
     *
     */
    protected void initSubscriber() {
        if (commType == null || commType.equals(""))
            commType = properties.getProperty("nb.comm.type", "niotcp");
        SessionService.setServiceConfigurationLocation(nbServConfPath);
        try {
            Properties props = new Properties();
            props.put("hostname", nbHost);
            props.put("portnum", nbPort);
            getEntityId();

            clientService = SessionService.getClientService(entityId);
            clientService.initializeBrokerCommunications(props, commType);

            eventConsumer = clientService.createEventConsumer(this);

            profile = clientService.createProfile(
                    TemplateProfileAndSynopsisTypes.STRING, this.subTopic);
            eventConsumer.subscribeTo(profile);
        } catch (ServiceException e) {
            e.printStackTrace();
            System.err.println(
                    "\nNaradaBrokering connection attempt failed... Terminating");
            System.exit(0);
        }
    }


    /**
     *
     *
     *
     */
    protected void initPublisher() {
        if (commType == null || commType.equals(""))
            commType = properties.getProperty("nb.comm.type", "niotcp");

        SessionService.setServiceConfigurationLocation(nbServConfPath);
        try {
            Properties props = new Properties();
            props.put("hostname", nbHost);
            props.put("portnum", nbPort);
            getEntityId();
            clientService = SessionService.getClientService(entityId);
            clientService.initializeBrokerCommunications(props, commType);

            eventProducer = clientService.createEventProducer();
            eventProducer.setSuppressRedistributionToSource(true);
            eventProducer.generateEventIdentifier(false);
            eventProducer.setDisableTimestamp(true);
            eventProducer.setContentSynopsisInfo(
                    TemplateProfileAndSynopsisTypes.STRING, this.pubTopic);
        } catch (ServiceException e) {
            e.printStackTrace();
            System.err.println(
                    "\nNaradaBrokering connection attempt failed... Terminating");
            System.exit(0);
        }
    }

    public boolean hasEventsToBePublished() {
        try {
            return eventProducer.hasEventsToBePublished();
        } catch (Exception ex) {
            return false;
        }
    }

    public boolean hasPendingTransfers() {
        try {
            return clientService.hasPendingTransfers();
        } catch (ServiceException ex) {
            return false;
        }
    }


    /**
     *
     *
     *
     * @param data byte[]
     */

    public void publishData(byte[] data) {
        NBEvent nbEvent;
        try {
            if (eventProducer != null && data != null) {
                nbEvent = eventProducer.generateEvent(data);
                eventProducer.publishEvent(nbEvent);
            }
        } catch (ServiceException e) {
            e.printStackTrace();
        }
    }

    /**
     *
     *
     *
     * @param data byte[]
     */

    public void publishDataWithProperty(byte[] data, String propertyName,
                                        String propertyValue) {
        NBEvent nbEvent;
        try {
            if (eventProducer != null && data != null) {
                nbEvent = eventProducer.generateEvent(data);
                nbEvent.getEventProperties().setMutableProperty(propertyName,
                        propertyValue, entityObj);
                eventProducer.publishEvent(nbEvent);
            }
        } catch (ServiceException e) {
            e.printStackTrace();
        } catch (NBEventException e) {
            e.printStackTrace();
        }
    }

    public NBEvent setEventPropertyValue(NBEvent nbEvent, String propertyName, String propertyValue) {
        try {
            nbEvent.getEventProperties().setMutableProperty(propertyName,
                    propertyValue, entityObj);
            return nbEvent;
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;
    }


    public String getEventPropertyValue(NBEvent nbEvent, String propertyName) {
        try {

            if (nbEvent.getEventProperties().hasProperties()) {
                if (nbEvent.getEventProperties().getProperty(propertyName) != null) {
                    String value = (String) nbEvent.getEventProperties().
                                   getProperty(propertyName);
                    return value;
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;
    }

    protected void startPublish() {
        NBEvent nbEvent;
        try {
            if (eventProducer != null) {
                nbEvent = eventProducer.generateEvent("".getBytes());
                nbEvent.getEventProperties().setImmutableProperty(
                        "MARKER",
                        "SOF", entityObj);
                eventProducer.publishEvent(nbEvent);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected void endPublish() {
        NBEvent nbEvent;
        try {
            while (true) {
                System.out.println("eventProducer.hasEventsToBePublished()=" +
                                   eventProducer.hasEventsToBePublished());
                if (!eventProducer.hasEventsToBePublished())
                    break;
            }
            if (eventProducer != null) {
                nbEvent = eventProducer.generateEvent("".getBytes());
                nbEvent.getEventProperties().setImmutableProperty(
                        "MARKER",
                        "EOF", entityObj);
                eventProducer.publishEvent(nbEvent);

            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }


    /**
     *
     * @param uid String
     * @throws ServiceException
     */
    public void closeBrokerConnection(String id) throws ServiceException {
        System.out.println("\nCLOSE CONNECTION\n");
        System.out.println("Received ID = " + id);
        System.out.println("\nHASH KEYS");
        Enumeration en = hash.keys();

        while (en.hasMoreElements()) {
            String item = (String) en.nextElement();
            if (item.equals(id)) {
                Filter f = (Filter) hash.get(id);
                if (f != null) {
                    f.terminateConnection(f);
                }
                break;
            }
        }
    }

    /**
     *
     *
     */
    protected void terminateConnection(Filter ff) {
        try {
            System.out.println("Terminating connection");
            if (ff.eventConsumer != null)
                ff.eventConsumer.unSubscribe(ff.profile);
            if (ff.eventProducer != null)
                ff.eventProducer.close();
            if (ff.getClientService() != null) {
                ff.getClientService().closeBrokerConnection();
                ff.getClientService().terminateServices();
            }
            ff.getHash().remove(ff.getID());
            System.out.println("HASH SIZE " + ff.getHash().size());
        } catch (ServiceException ex) {
        }
    }


    public Hashtable getHash() {
        return hash;
    }

    public String getID() {
        return uid;
    }

//    public String getPubTopic() {
//        return pubTopic;
//    }
//
//    public String getSubTopic() {
//        return subTopic;
//    }
//
//    public String getNbHost() {
//        return nbHost;
//    }
//
//    public String getNbPort() {
//        return nbPort;
//    }
//
//    public String getCommType() {
//        return commType;
//    }
//
//    public void setPubTopic(String pubTopic) {
//        this.pubTopic = pubTopic;
//    }
//
//    public void setSubTopic(String subTopic) {
//        this.subTopic = subTopic;
//    }
//
//    public void setNbHost(String nbHost) {
//        this.nbHost = nbHost;
//    }
//
//    public void setNbPort(String nbPort) {
//        this.nbPort = nbPort;
//    }
//
//    public void setCommType(String commType) {
//        this.commType = commType;
//    }

    public ClientService getClientService() {
        return clientService;
    }


    private Class currentFilter;
    private Object filterInstance;
//     private String uid;
//     private UUIDGenerator uig = UUIDGenerator.getInstance();
//     private static Hashtable hash = new Hashtable();


    /**
     *
     * @param filterName String
     * @param args String[]
     * @param filterSecificArgs String[]
     * @return String
     */
    public String startFilter(String filterName,
                              String[] args,
                              String[] filterSecificArgs) {
        if (args == null || args.length == 0 || args.length < 5) {
            System.err.println("Incorrect number of arguments");
            return "";
        }
        try {
            currentFilter = Class.forName(filterName);
            Class[] a = new Class[] {String[].class, String[].class};
            Constructor c = currentFilter.getConstructor(a);
            Object[] o = new Object[] {args, filterSecificArgs};
            filterInstance = c.newInstance(o);

            Method m = currentFilter.getMethod("getID", null);
            String id = (String) m.invoke(filterInstance, null);
//             hash.put(id, this);
            System.out.println("START FILTER id = " + id);
            return id;
        } catch (Exception ex2) {
            ex2.printStackTrace();
            return "";
        }
    }

    /**
     *
     * @param filterName String
     * @param filterID String
     */
    public void stopFilter(String filterName,
                           String filterID) {
        try {

            Filter f = (Filter) hash.get(filterID);

            currentFilter = Class.forName(filterName);
            Class[] a = new Class[] {};
            //Constructor c = currentFilter.getConstructor(a);
            Object[] o = new Object[] {};
            //filterInstance = c.newInstance(o);
            a = new Class[] {String.class};
            Method m = currentFilter.getMethod("closeBrokerConnection", a);
            o = new Object[] {filterID};
            m.invoke(this, o);
        } catch (Exception ex2) {
            ex2.printStackTrace();
        }
    }


    /**
     *
     * @param args String[]
     */
    public static void main(String[] args) {
        try {
            String site = "PIN2";
            String[] argsa = {"niotcp",
                             "gf2.ucs.indiana.edu",
                             "3045",
                             "/SOPAC/GPS/OCRTN/RYO",
                             "/SOPAC/GPS/OCRTN/POS"};
            String[] argsb = {site};

//        Filter f = Filter.getInstance();
//        Filter f1 = Filter.getInstance();
//
//        String id = f.startFilter(
//                "cgl.sensorgrid.sopac.gps.filters.ryo2pos", argsa,
//                argsb);
//
//        String id1 = f1.startFilter(
//                "cgl.sensorgrid.sopac.gps.filters.ryo2pos", argsa,
//                argsb);
//
//
//        Thread.sleep(8000);
//        f.stopFilter("cgl.sensorgrid.sopac.gps.filters.ryo2pos",
//                     id1);
//
//
//        Thread.sleep(8000);
//        System.out.println("CLOSING F2");
//        Thread.sleep(1000);
//
//        f.stopFilter("cgl.sensorgrid.sopac.gps.filters.ryo2pos",
//                     id);
//
//
//
//
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

}
